package com.by.order.service.impl;

import com.by.common.constant.OrderConstant;
import com.by.common.exception.ErrorCode;
import com.by.common.redis.RedisKeyUtils;
import com.by.order.api.service.IOrderService;
import com.by.order.api.service.IReliableMessageService;
import com.by.order.api.vo.OrderVo;
import com.by.order.entity.OrderEntity;
import com.by.order.mapper.OrderMapper;
import com.by.order.service.RedisRateLimiter;
import com.by.product.api.service.IProductService;
import com.by.product.api.vo.ProductVo;
import com.by.rpc.annotation.RpcReference;
import com.by.rpc.annotation.RpcService;
import com.by.rpc.exception.BusinessException;
import com.by.rpc.util.RedisLuaHelper;
import com.by.user.api.service.IUserService;
import com.by.user.api.vo.UserVo;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;

@Slf4j
@Service
@RpcService
public class OrderServiceImpl implements IOrderService {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private RedisRateLimiter redisRateLimiter;

    @Autowired
    private RedisLuaHelper redisLuaHelper;

    @Autowired
    private SlidingWindowRateLimiter slidingWindowRateLimiter;

    @Autowired
    private SnowflakeIdGenerator snowflakeIdGenerator;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @RpcReference(interfaceClass = IProductService.class, version = "1.0.0", group = "default")
    private IProductService productService;

    @RpcReference(interfaceClass = IUserService.class, version = "1.0.0", group = "default")
    private IUserService userService;

    @Autowired
    private IReliableMessageService reliableMessageService;

    // 修复：Lua脚本只做库存预扣，不再处理用户去重。
    // 因为用户去重状态必须和本地事务绑定。
    private static final String SECKILL_LUA_SCRIPT = """
                -- KEYS[1]: stockKey
                -- 1. 检查库存
                local stockKey = KEYS[1]
                local stock = tonumber(redis.call('get', stockKey))
                if not stock or stock <= 0 then
                    return 0
                end
                -- 2. 扣减库存
                redis.call('decr', stockKey)
                return 1
            """;

    private DefaultRedisScript<Long> seckillRedisScript;

    @PostConstruct
    public void init() {
        seckillRedisScript = new DefaultRedisScript<>();
        seckillRedisScript.setScriptText(SECKILL_LUA_SCRIPT);
        seckillRedisScript.setResultType(Long.class);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public OrderVo seckillCreateOrder(Long userId, Long goodsId) {
        // 1. 参数和限流校验
        if (userId == null || goodsId == null) {
            throw new BusinessException(ErrorCode.PARAM_INVALID.getCode(), "用户ID和商品ID不能为空");
        }

        // 用户级滑动窗口限流
        String userSeckillLimitKey = RedisKeyUtils.getUserSeckillLimitKey(userId, goodsId);
        if (!slidingWindowRateLimiter.tryAcquire(userSeckillLimitKey, 10, 3)) {
            throw new BusinessException(ErrorCode.SECKILL_FREQUENT.getCode(), "您的请求过于频繁，请稍后再试");
        }

        // 全局秒杀限流（带降级策略）
        String globalSeckillKey = RedisKeyUtils.getSeckillRateLimitKey(goodsId);
        if (!tryAcquireWithFallback(globalSeckillKey, 500, 1)) {
            throw new BusinessException(ErrorCode.SECKILL_FREQUENT.getCode(), "系统繁忙，请稍后再试");
        }

        // 修复：在执行Lua脚本前，先检查用户是否已经参与过秒杀。
        // 这是最关键的改动，防止重复下单，同时确保这个检查是事务无关的。
        String userSetKey = RedisKeyUtils.getSeckillUserSetKey(goodsId);
        if (Boolean.TRUE.equals(redisTemplate.opsForSet().isMember(userSetKey, String.valueOf(userId)))) {
            throw new BusinessException(ErrorCode.SECKILL_DUPLICATE.getCode(), "您已参与过该商品秒杀");
        }

        // 2. 执行秒杀核心逻辑（Lua脚本只做库存预扣）
        String stockKey = RedisKeyUtils.getSeckillStockKey(goodsId);
        Object luaResult = redisLuaHelper.executeStringLua(seckillRedisScript, List.of(stockKey));

        if (!(luaResult instanceof Long)) {
            log.error("秒杀脚本返回非预期结果:{}，商品ID:{}，用户ID:{}", luaResult, goodsId, userId);
            throw new BusinessException(ErrorCode.SECKILL_SOLD_OUT.getCode(), "秒杀过程异常，请稍后再试");
        }

        int code = ((Long) luaResult).intValue();
        if (code != 1) {
            // 此时，code 只能是 0，表示商品已售罄
            throw new BusinessException(ErrorCode.SECKILL_SOLD_OUT.getCode(), "商品已售罄");
        }

        // 3. 核心业务流程，全部包裹在 try...catch 中，确保 Redis 预扣库存能被回滚
        try {
            // 4. 检查用户和商品信息（使用RPC）
            UserVo user = userService.getUserById(userId);
            if (user == null) {
                throw new BusinessException(ErrorCode.USER_NOT_EXIST.getCode(), "用户不存在");
            }
            ProductVo product = productService.getProductById(goodsId);
            if (product == null) {
                throw new BusinessException(ErrorCode.GOODS_NOT_EXIST.getCode(), "商品不存在");
            }

            // 5. 创建订单实体
            OrderEntity orderEntity = new OrderEntity();
            orderEntity.setId(snowflakeIdGenerator.nextId());
            orderEntity.setUserId(userId);
            orderEntity.setGoodsId(goodsId);
            orderEntity.setGoodsName(product.getName());
            orderEntity.setGoodsCount(1);
            orderEntity.setGoodsPrice(product.getPrice());
            orderEntity.setOrderNo(generateOrderNo());
            orderEntity.setStatus(0); // 0:待支付
            orderEntity.setCreateTime(LocalDateTime.now());
            orderEntity.setExpireTime(LocalDateTime.now().plusMinutes(OrderConstant.ORDER_EXPIRE_TIME));

            // 6. 在本地事务中，同时插入订单记录和可靠消息记录
            orderMapper.insert(orderEntity);
            // 调用我们重构后的可靠消息服务方法
            reliableMessageService.recordMessageInLocalTx(
                    "商品数据库库存扣减",
                    String.valueOf(orderEntity.getId()),
                    OrderConstant.MESSAGE_TYPE_STOCK_DECREASE
            );

            // 修复：只有当整个本地事务成功提交后，才将用户ID添加到去重集合
            // 这样就保证了 Redis 的用户去重状态和数据库中的订单记录一致
            redisTemplate.opsForSet().add(userSetKey, String.valueOf(userId));

            log.info("秒杀订单创建成功并已记录，订单ID:{}，商品ID:{}，用户ID:{}",
                    orderEntity.getId(), goodsId, userId);

            // 7. 构造并返回结果
            OrderVo orderVo = new OrderVo();
            BeanUtils.copyProperties(orderEntity, orderVo);
            return orderVo;
        } catch (Exception e) {
            log.warn("检测到秒杀业务异常，回滚Redis库存。商品ID:{}，用户ID:{}，异常信息:{}",
                    goodsId, userId, e.getMessage());

            // 修复：此时只需要回滚预扣的 Redis 库存
            // 因为用户去重状态根本没有被设置，所以无需回滚
            redisTemplate.opsForValue().increment(stockKey, 1);

            // 重新抛出异常，让上层调用者和 Spring 事务管理器感知到业务失败，进行事务回滚
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            } else {
                throw new RuntimeException("秒杀过程发生非预期异常，已回滚", e);
            }
        }
    }

    // --- 其他方法 ---

    @Override
    public OrderVo getOrderById(Long id) {
        OrderEntity orderEntity = orderMapper.selectById(id);
        if (orderEntity == null) {
            return null;
        }

        OrderVo orderVo = new OrderVo();
        BeanUtils.copyProperties(orderEntity, orderVo);
        return orderVo;
    }

    @Override
    @Transactional
    public boolean payOrder(Long id) {
        OrderEntity orderEntity = orderMapper.selectById(id);
        if (orderEntity == null) {
            return false;
        }

        if (orderEntity.getStatus() != 0) {
            return false;
        }

        if (orderEntity.getExpireTime().isBefore(LocalDateTime.now())) {
            return false;
        }

        orderEntity.setStatus(1); // 1:已支付
        orderEntity.setPayTime(LocalDateTime.now());
        orderMapper.updateById(orderEntity);

        // 支付成功后发送消息，通知商品服务确认库存扣减（或扣款）
        reliableMessageService.recordMessageInLocalTx(
                "订单支付成功，订单ID: " + id,
                String.valueOf(id),
                OrderConstant.MESSAGE_TYPE_STOCK_CONFIRM
        );

        return true;
    }

    @Override
    @Transactional
    public boolean cancelOrder(Long id) {
        OrderEntity orderEntity = orderMapper.selectById(id);
        if (orderEntity == null) {
            return false;
        }

        if (orderEntity.getStatus() != 0) {
            return false;
        }

        orderEntity.setStatus(2); // 2:已取消
        orderEntity.setCancelTime(LocalDateTime.now());
        orderMapper.updateById(orderEntity);

        // 发送消息通知商品服务回滚库存
        reliableMessageService.recordMessageInLocalTx(
                "订单已取消，回滚库存，订单ID: " + id,
                String.valueOf(id),
                OrderConstant.MESSAGE_TYPE_STOCK_ROLLBACK
        );

        return true;
    }

    /**
     * 尝试获取令牌，带降级策略
     *
     * @return 是否获取成功
     */
    private boolean tryAcquireWithFallback(String key, int windowInSeconds, int limit) {
        try {
            return redisRateLimiter.tryAcquire(key, windowInSeconds, limit);
        } catch (Exception e) {
            log.error("全局限流器执行异常，采取安全降级策略：拒绝请求，key: {}", key, e);
            return false;
        }
    }

    /**
     * 生成订单号
     */
    private String generateOrderNo() {
        return "ORDER-" + snowflakeIdGenerator.nextId();
    }
}